package main

import (
	"fmt"
	"log"
	"saasems/config"
	"saasems/database"
	"saasems/migrations"

	"saasems/kafka"

	_ "github.com/lib/pq"
)

func main() {
	// 加载配置文件
	cfg, err := config.LoadConfig()
	if err != nil {
		log.Fatalf("配置文件加载失败: %v", err)
	}
	fmt.Println("Starting IoT SaaS Platform...")

	// 初始化数据库连接
	postgresDB, err := database.NewPostgresDB(fmt.Sprintf(
		"postgres://%s:%s@%s:%d/%s?sslmode=%s",
		cfg.Postgres.User,
		cfg.Postgres.Password,
		cfg.Postgres.Host,
		cfg.Postgres.Port,
		cfg.Postgres.DBName,
		cfg.Postgres.SSLMode,
	))
	if err != nil {
		log.Fatalf("无法连接PostgreSQL数据库: %v", err)
	}
	influxDB, err := database.NewInfluxDB(cfg.InfluxDB.URL, cfg.InfluxDB.Org)
	if err != nil {
		log.Fatalf("无法连接Influxdb数据库：%v", err)
	}
	defer postgresDB.Close()
	defer influxDB.Close()

	// Get the GORM DB instance from PostgresDB
	gormDB := postgresDB.GetDB()
	migrations.CreateDbTable(gormDB)

	// 初始化MQTT连接
	mqttClient := connectMQTT(cfg.MQTT.Broker, cfg.MQTT.ClientID, cfg.MQTT.Username, cfg.MQTT.Password)
	defer mqttClient.Disconnect(250)

	// 修改数据库连接为工厂函数
	influxDBFactory := func(tenantID string) *database.InfluxDB {
		// 根据租户ID返回对应的InfluxDB连接
		db, err := database.NewInfluxDB(cfg.InfluxDB.URL, tenantID)
		if err != nil {
			log.Fatalf("无法创建InfluxDB连接: %v", err)
		}
		return db
	}

	postgresDBFactory := func(tenantID string) *database.PostgresDB {
		// 根据租户ID返回对应的PostgresDB连接
		db, err := database.NewPostgresDB(fmt.Sprintf(
			"postgres://%s:%s@%s:%d/%s_%s?sslmode=%s",
			cfg.Postgres.User,
			cfg.Postgres.Password,
			cfg.Postgres.Host,
			cfg.Postgres.Port,
			cfg.Postgres.DBName,
			tenantID,
			cfg.Postgres.SSLMode,
		))
		if err != nil {
			log.Fatalf("无法创建PostgresDB连接: %v", err)
		}
		return db
	}

	// 初始化Kafka处理器

	//kafkaProcessor, err := NewKafkaProcessor(
	//	cfg.Kafka.Brokers,
	//	cfg.Kafka.GroupID,
	//	influxDBFactory,
	//	postgresDBFactory,
	//)
	//if err != nil {
	//	log.Fatalf("Failed to create Kafka processor: %v", err)
	//}
	//
	//// 启动Kafka消息处理
	//go func() {
	//	kafkaProcessor.ProcessMessages([]string{"raw-device-data"})
	//}()

	processor := kafka.NewKafkaProcessor(
		[]string{"localhost:9092"},
		"mqtt-data",
		"processed-data",
		postgresDBFactory,
		influxDBFactory,
	)
	go processor.Start()

	// 主程序保持运行
	select {}
}
